package com.lagou.rpc.client.protocol;

import com.lagou.rpc.client.LoadBalance;
import com.lagou.rpc.client.RoundRobinLoadBalance;
import com.lagou.rpc.client.RpcFuture;
import com.lagou.rpc.client.netty.NettyClient;
import com.lagou.rpc.entity.RpcRequest;
import com.lagou.rpc.zk.ChildListener;
import com.lagou.rpc.zk.ZkClient;
import org.apache.curator.framework.api.CuratorWatcher;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.stream.Collectors;

@Component
public class RpcInvoker {

    @Autowired
    ZkClient zkClient;

    // 每台服务端和netty客户端连接的对应关系
    private ConcurrentMap<String, NettyClient> urlAndClient = new ConcurrentHashMap<>();

    // service和child文件夹的对应关系
    private ConcurrentMap<String, List<String>> serviceChild = new ConcurrentHashMap<>();

    //已注册过监听的watcher的zookeeper路径
    private Set<String> watcherPath = Collections.newSetFromMap(new ConcurrentHashMap<>());
    private LoadBalance loadBalance;
    private ScheduledExecutorService threadPool = Executors.newScheduledThreadPool(8);
    private Logger log = LoggerFactory.getLogger(getClass());

    public RpcInvoker() {

    }

    @PostConstruct
    void init() {

        this.loadBalance = new RoundRobinLoadBalance(zkClient);
        //全量监听
        serviceList().forEach(service -> {
            List<String> child = zkClient.getChildren("/rpc/" + service);
            notify(service, child);
        });

        serviceList().forEach(service -> {
            String path = "/rpc/" + service;
            if (this.watcherPath.add(path)) {
                CuratorWatcher watcher = zkClient.createChildListener(path, new ChildListener() {
                    @Override
                    public void childChanged(String path, List<String> child) {
                        RpcInvoker.this.notify(service, child);
                    }
                });
                zkClient.addChildListener(path, watcher);
            }
        });

    }

    void notify(String service, List<String> child) {
        List<String> validUrl = child.stream().filter(url -> {
            return initClient(url) != null;
        }).collect(Collectors.toList());

        // 这个validUrl,好像不太行。。。
        serviceChild.put(service, child);
    }

    /**
     * 初始化nettyclient,可自动重连
     * @param url
     * @return
     */
    NettyClient initClient(String url) {
        NettyClient client = urlAndClient.get(url);
        if (client != null) {
            if (!client.isClosed()) {
                return client;
            } else {
                client.close();
                log.info("服务端关闭了连接:" + url);
                urlAndClient.remove(url);
            }
        }
        synchronized (this) {
            if (urlAndClient.containsKey(url)) {
                return urlAndClient.get(url);
            }

            client = new NettyClient(url);
            if (!client.hasChannel()) {
                log.info("服务端没有连接成功:" + url);
                return null;
            }
            return urlAndClient.put(url, client);
        }
    }

    static List<String> serviceList() {
        return Arrays.asList("userService", "resumeService");
    }

    /**
     * 创建一个代理类。
     * @param interfaceClass
     * @param <T>
     * @return
     * @throws Exception
     */
    public <T> T createProxy(Class interfaceClass) throws Exception {
        String service = lowerFirst(interfaceClass.getSimpleName());
        String url = "/rpc/" + service + ".lastTime";
        List<String> child = this.serviceChild.get(service);
        String select = loadBalance.doSelect(url, child);
        NettyClient client = urlAndClient.get(select);
        return (T) Proxy.newProxyInstance(
                Thread.currentThread().getContextClassLoader(),
                new Class<?>[]{interfaceClass},
                handler(service, client));
    }

    InvocationHandler handler(String serviceName, NettyClient client) {
        return new InvocationHandler() {
            @Override
            public Object invoke(Object o, Method method, Object[] objects) throws Throwable {
                RpcRequest rpcRequest = new RpcRequest();
                rpcRequest.setClassName(serviceName);
                rpcRequest.setRequestId(UUID.randomUUID().toString());
                rpcRequest.setMethodName(method.getName());
                rpcRequest.setParameters(objects);
                if (client == null) {
                    return "客户端调用失败了.";
                }

                RpcFuture future = client.getClientHandler().sendRequest(rpcRequest);
                return future.get();
            }
        };
    }

    String lowerFirst(String name) {
        if (name == null) {
            return "";
        }
        char[] arr = name.toCharArray();
        arr[0] += 32;
        return new String(arr);
    }
}
